[DRAFT] update evaluate to be concurrent #1345
Conversation
| @@ -642,6 +651,61 @@ async def astart(self) -> _AsyncExperimentManager: | |||
| upload_results=self._upload_results, | |||
| ) | |||
|
|
|||
| async def awith_predictions_and_evaluators( | |||
There was a problem hiding this comment.
We could probably do something similar to what we do in the sync version to avoid having to duplicate logic here (basically share a semaphor)
| evaluators = _resolve_evaluators(evaluators) | ||
|
|
||
| if not hasattr(self, "_evaluator_executor"): | ||
| self._evaluator_executor = cf.ThreadPoolExecutor(max_workers=4) |
There was a problem hiding this comment.
ooc where's the 4 come from?
There was a problem hiding this comment.
I copied the value from _ascore - not really sure beyond that
| ) | ||
| async with lock: |
There was a problem hiding this comment.
could we just return the selected_results in _run_single_evaluator and construct the eval_results after the asycio.gather? to avoid needing to lock?
There was a problem hiding this comment.
should be fixed, but someone should check I did it correctly
| { | ||
| name: { | ||
| "presigned_url": value["presigned_url"], | ||
| "reader": io.BytesIO(value["reader"].getvalue()), |
There was a problem hiding this comment.
would love @agola11's input on this bit
| new_attachments[name] = { | ||
| "presigned_url": attachment["presigned_url"], | ||
| "reader": io.BytesIO( | ||
| self._attachment_raw_data_dict[str(example.id) + name] |
There was a problem hiding this comment.
you're sure this doesn't copy the bytes?
There was a problem hiding this comment.
No, you are correct. io.BytesIO copies the underlying bytes. This is wrong, I am working on a fix rn.
There was a problem hiding this comment.
ehh actually I am going to walk back my statement. based on testing I don't think bytesIO copies the data.
No description provided.